package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
/**
 * ClassName: DwsLearnChapterPlayWindow
 * Description:
 * 统计各窗口各章节播放次数、播放总时长、播放独立用户数 !!!
 * 统计各窗口 各章节 的播放次数 + 播放总时长 + 播放独立用户数
 */

public class DwsLearnChapterPlayWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsLearnChapterPlayWindow().start(10024,4,"dws_learn_chapter_play_window", Constant.TOPIC_DWD_LEARN_PLAY);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS) {
        //TODO 1.jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.map(JSON::parseObject);

        //TODO 2.按照userId分组
        KeyedStream<JSONObject, String> userIdKeyedDS = jsonObjDS.keyBy(j -> j.getString("userId"));

        //TODO 3.使用flink状态编程，转换为实体类对象
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> beanDS = userIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, DwsLearnChapterPlayWindowBean>() {

                    private ValueState<String> userLastPlayDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateProperties =
                                new ValueStateDescriptor<String>("userLastPlayDateState", String.class);
                        stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        userLastPlayDateState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj,
                                               KeyedProcessFunction<String, JSONObject, DwsLearnChapterPlayWindowBean>.Context ctx,
                                               Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {

                        //获取当前时间
                        Long ts = jsonObj.getLong("ts");
                        String curDate = DateFormatUtil.tsToDate(ts);

                        //获取视频ID
                        String videoId = jsonObj.getString("videoId");

                        //获取播放总时长
                        Long playSec = jsonObj.getLong("playSec");

                        //获取状态并赋值观看人数
                        long uuCt = 0L;
                        String lastDate = userLastPlayDateState.value();
                        if (StringUtils.isEmpty(lastDate) || !lastDate.equals(curDate)) {
                            uuCt = 1L;
                            userLastPlayDateState.update(curDate);
                        }

                        out.collect(
                                DwsLearnChapterPlayWindowBean
                                        .builder()
                                        .videoId(videoId)
                                        .playCount(1L)
                                        .playTotalSec(playSec)
                                        .playUuCount(uuCt)
                                        .ts(ts)
                                        .build()
                        );

                    }
                }
        );
        //beanDS.print();


        //TODO 4.Watermark & 事件时间字段
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<DwsLearnChapterPlayWindowBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsLearnChapterPlayWindowBean>() {
                                    @Override
                                    public long extractTimestamp(DwsLearnChapterPlayWindowBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );


        //TODO 5.按照维度视频ID分组
        KeyedStream<DwsLearnChapterPlayWindowBean, String> videoKeyedDS = withWatermarkDS.keyBy(DwsLearnChapterPlayWindowBean::getVideoId);

        //TODO 6.开窗
        WindowedStream<DwsLearnChapterPlayWindowBean, String, TimeWindow> windowDS =
                videoKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 7.聚合
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> reduceDS = windowDS.reduce(
                new ReduceFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public DwsLearnChapterPlayWindowBean reduce(DwsLearnChapterPlayWindowBean value1, DwsLearnChapterPlayWindowBean value2) throws Exception {
                        value1.setPlayCount(value1.getPlayCount() + value2.getPlayCount());
                        value1.setPlayTotalSec(value1.getPlayTotalSec() + value2.getPlayTotalSec());
                        value1.setPlayUuCount(value1.getPlayUuCount() + value2.getPlayUuCount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        ProcessWindowFunction<DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean, String, TimeWindow>.Context context,
                                        Iterable<DwsLearnChapterPlayWindowBean> elements,
                                        Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                        DwsLearnChapterPlayWindowBean bean = elements.iterator().next();

                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());

                        bean.setStt(stt);
                        bean.setEdt(edt);
                        bean.setCurDate(curDate);

                        out.collect(bean);
                    }
                }
        );
        //reduceDS.print();

        //TODO 8.关联视频维度表
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withChapterIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public void addDims(DwsLearnChapterPlayWindowBean bean, JSONObject dimJsonObj) {
                        bean.setChapterId(dimJsonObj.getString("chapter_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getRowKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getVideoId();
                    }
                },
                60L,
                TimeUnit.SECONDS
        );

        //TODO 9.关联章节维度表
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withChapterNameDS = AsyncDataStream.unorderedWait(
                withChapterIdDS,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public void addDims(DwsLearnChapterPlayWindowBean bean, JSONObject dimJsonObj) {
                        bean.setChapterName(dimJsonObj.getString("chapter_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_chapter_info";
                    }

                    @Override
                    public String getRowKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getChapterId();
                    }
                },
                60L,
                TimeUnit.SECONDS
        );

        //withChapterNameDS.print();

        //TODO 10.写到Doris
        withChapterNameDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_learn_chapter_play_window"));

    }
}
